/** @file    BusManager.cpp
 *  @author  Alessandro Polo
 *  @version $Id: BusManager.cpp 2874 2010-08-08 19:12:58Z alex $
 *  @brief
 * File containing methods for the wosh::BusManager class.
 * The header for this class can be found in BusManager.h, check that file
 * for class description.
 ****************************************************************************/
/* Copyright (c) 2007-2010, WOSH - Wide Open Smart Home 
 * by Alessandro Polo - OpenSmartHome.com
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *     * Redistributions of source code must retain the above copyright
 *       notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above copyright
 *       notice, this list of conditions and the following disclaimer in the
 *       documentation and/or other materials provided with the distribution.
 *     * Neither the name of the OpenSmartHome.com WOSH nor the
 *       names of its contributors may be used to endorse or promote products
 *       derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY Alessandro Polo ''AS IS'' AND ANY
 * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL Alessandro Polo BE LIABLE FOR ANY
 * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 ****************************************************************************/
 
 #include <core/BusManager.h>
 #include <core/ObjectFactory.h>
 #include <core/List.h>
 #include <core/Table.h>
 #include <core/MethodsCommon.h>
 #include <core/Utilities.h>
 #include <core/NetworkManager.h>
 #include <core/SystemError.h>
 #include <core/WoshKernel.h>
 #include <core/DataModels.h>
 #include <core/BusRing.h> //it's default bus-type


using namespace std;
namespace wosh {

// Gamma singleton! static reference is allocated and maintained by WoshKernel
extern "C" {
	BusManager* BusManager::getInstance() { 
		if ( WoshKernel::getInstance() == NULL ) return NULL;
		return &WoshKernel::getInstance()->busses();
	}
}
 // register as abstract because it is a singleton:
 WOSH_REGISTER_INTERFACE(wosh::BusManager, "wosh::WoshModule", 0.01, _static_BusManager )

//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////// CONSTRUCTORS

BusManager::BusManager( const WoshKernel* parent )
	: WoshModule(parent, _BusManager_NAME, _BusManager_URI) {

//Log.setLevel( VERBOSE );

	Log(LOG_DEBUG, " Registering Methods.." );
	MethodRequest* mmCreateBus = Methods.create<MethodRequest>( _BusManager_METHOD_create_bus, "create new Bus", Permission::R_XR__R__() );
	mmCreateBus->setMethod( this, (MethodRequestPtr)&BusManager::mmDoCreateBus );

	MethodRequest* mmControlBus = Methods.create<MethodRequest>( _BusManager_METHOD_control_bus, "control Bus", Permission::R_XR__R__() );
	mmControlBus->setMethod( this, (MethodRequestPtr)&BusManager::mmDoControlBus );

	MethodList* mmListBusses = Methods.create<MethodList>( _METHOD_List, "List Busses", Permission::R_XR_XR__() );
	mmListBusses->setDataFieldExtractor( DataFieldExtractor<Bus>::createFor< DataAdapterMapT<string, Bus> >(this->busses, "Busses") );

 }

BusManager::~BusManager() {
	Log(LOG_DEBUG, ":~BusManager() Destroying.." );

	if ( this->busses.count() != 0 ) {
		Log(LOG_VERBOSE, ":~BusManager() : Freeing %d Busses..", this->busses.count() );
		stopBusses(20000);
	 }

	vector<Bus*> instances;
	this->busses.transactionBeginWrite();
	tStrBusMap::Iterator it;
	const tStrBusMap::ConstIterator it_end = this->busses.getMap().end();
	for ( it=this->busses.getMap().begin(); it!=it_end; ++it ) {
		if ( it->second == NULL ) continue;
		instances.push_back(it->second);
	 }
	this->busses.getMap().clear();
	this->busses.transactionEnd();

	vector<Bus*>::iterator iit;
	for ( iit=instances.begin(); iit!=instances.end(); ++iit ) {
		if ( *iit == NULL ) continue;
		delete *iit;
	 }
	this->busses.clear();

	Log(LOG_VERBOSE, " Destroyed." );
 }

//////////////////////////////////////////////////////////////////////////////////////////////// CONSTRUCTORS
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
/////////////////////////////////////////////////////////////////////////////////////////////// KERNEL EVENTS

WRESULT BusManager::onKernelInitializing() {
	WRESULT ret = WoshModule::onKernelInitializing();
	return ret;
 }

WRESULT BusManager::onKernelStopped() {
	Log(LOG_DEBUG, ":onKernelStopped() Stopping Busses.." );
	WRESULT ret = stopBusses(30000);
	if ( WFAILED(ret) )
		Log(LOG_CRITICAL, ":onKernelStopping() ERROR#%d Stopping Busses.", ret );
	ret += WoshModule::onKernelStopped();
	return ret;
 }

/////////////////////////////////////////////////////////////////////////////////////////////// KERNEL EVENTS
//////////////////////////////////////////////////////////////////////////////////////////////////////////////

int BusManager::countRunning() const {
	MutexLockerRead mL(this->busses.getMutex());
	return countRunning_();
 }

int BusManager::countRunning_() const {
	int ret = 0;
	tStrBusMap::ConstIterator it;
	const tStrBusMap::ConstIterator it_end = this->busses.end();
	for ( it=this->busses.begin(); it!=it_end; ++it ) {
		if ( it->second == NULL ) continue;
		if ( it->second->isBusRunning() )
			ret++;
	 }
	return ret;
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////

bool BusManager::existsBus( const std::string& busname ) const {
	MutexLockerRead mL(this->busses.getMutex());
	return this->busses.exists( busname );
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////////////////////

WRESULT BusManager::registerBus( Bus* bus_instance ) {
	if ( bus_instance == NULL ) return WRET_ERR_PARAM;
	MutexLockerWrite mL(this->busses.getMutex());
	return this->registerBus_( bus_instance );
 }

WRESULT BusManager::registerBus_( Bus* bus ) {
	if ( bus == NULL ) return WRET_ERR_PARAM;
	if ( bus->getName() == "" ) return WRET_ERR_PARAM;

	if ( this->busses.exists(bus->getName()) ) {
		SystemError::raise(this, Log, SystemError::WARNING, 0, "registerBus", "WARNING#%d Bus %s exists!", WRET_ERR_PARAM, bus->getName().c_str() );
		return WRET_ERR_PARAM;
	 }
	if ( this->busses.existsValue(bus) ) {
		SystemError::raise(this, Log, SystemError::CRITICAL, 0, "registerBus", "FAILED#%d Bus INSTANCE of %s already registered (with different name!?)!", WRET_ERR_PARAM, bus->getName().c_str() );
		return WRET_ERR_PARAM;
	 }
	bus->setParentObject(this);
	this->busses.set( bus->getName(), bus );
	if ( !bus->isBusRunning() ) {
		Log(LOG_VERBOSE, ":registerBus(%s) Starting Bus#%ld", bus->getName().c_str(), bus->getBusID() );
		WRESULT ret = bus->startBus();
		if ( WFAILED(ret) )
			Log(LOG_CRITICAL, ":registerBus(%s) : FAIELD Starting Bus#%ld!", bus->getName().c_str(), bus->getBusID() );
		else
			Log(LOG_INFO, ":registerBus(%s) : Bus#%ld Ready (running)!", bus->getName().c_str(), bus->getBusID() );
	 }
	else
		Log(LOG_INFO, ":registerBus(%s) : Bus#%ld (al)Ready (running)!", bus->getName().c_str(), bus->getBusID() );

	return WRET_OK;
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////

WRESULT BusManager::unRegisterBus_( Bus* bus ) {
	if ( bus == NULL ) return WRET_ERR_PARAM;
	if ( bus->getName() == "" ) return WRET_ERR_PARAM;

	if ( !this->busses.exists(bus->getName()) ) {
		SystemError::raise(this, Log, SystemError::WARNING, 0, "unRegisterBus", "WARNING#%d Bus %s noy found!", WRET_ERR_PARAM, bus->getName().c_str() );
	 }
	if ( !this->busses.existsValue(bus) ) {
		SystemError::raise(this, Log, SystemError::CRITICAL, 0, "unRegisterBus", "FAILED#%d Bus INSTANCE not found!", WRET_ERR_PARAM, bus->getName().c_str() );
		return WRET_ERR_PARAM;
	 }

	this->busses.erase(bus->getName(), false);
	//Log(LOG_INFO, ":unRegisterBus() Bus '%s' unregistered", bus->getName().c_str() );
	return WRET_OK;
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////////////////////

WRESULT BusManager::connectBus( IBusListener* handler, const std::string& busname, int connection_options ) {
	if ( handler == NULL || busname.size() < 3 ) return WRET_ERR_PARAM;
	if ( connectBusEx(handler, busname, connection_options) == NULL )
		return WRET_ERR_INTERNAL;
	return WRET_OK;
 }

WRESULT BusManager::disconnectBus( IBusListener* handler, const std::string& busname, int connection_options ) {
	if ( handler == NULL ) return WRET_ERR_PARAM;
	bool deferred = ( (connection_options & (Bus::ACTION_DEFERRED)) == Bus::ACTION_DEFERRED);
	MutexLockerRead mL(this->busses.getMutex());
	if ( busname == "" || busname == "*" ) {
		Log(LOG_VERBOSE, ":disconnectBus(%p,%s,%i) : UnListening all busses", handler, busname.c_str(), connection_options );
		tStrBusMap::Iterator it;
		const tStrBusMap::ConstIterator it_end = this->busses.end();
		for ( it=this->busses.begin(); it!=it_end; ++it ) {
			if ( it->second == NULL ) continue;
			if ( it->second->isBusListener(handler) )
				it->second->removeBusListener(handler, deferred);
		 }
		return WRET_OK;
	 }
	Bus* bus = this->busses.find(busname);
	if ( bus != NULL ) {
		bus->removeBusListener(handler, deferred);
		Log(LOG_VERBOSE, ":disconnectBus(%p,%s,%i) : UnListening Bus#%ld", handler, busname.c_str(), connection_options, bus->getBusID() );
	 }
	else
		Log(LOG_WARNING, ":disconnectBus(%p,%s,%i) : FAILED: Bus not found!", handler, busname.c_str(), connection_options );
	return WRET_OK;
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////

Bus* BusManager::connectBusEx( IBusListener* handler, const std::string& busname, int connection_options ) {
	if ( handler == NULL ) return NULL;
	if ( busname.size() < 3 ) return NULL;
	this->busses.transactionBeginRead();
	Bus* bus = this->busses.find(busname);
	if ( bus == NULL ) {
		bool create_if_required = ( (connection_options & (Bus::BUS_CREATE_IF_MISSING)) == Bus::BUS_CREATE_IF_MISSING);
		if ( !create_if_required ) {
			this->busses.transactionEnd();
			Log(LOG_CRITICAL, ":connectBus(%p,%s,%i) : FAILED: Bus not found!", handler, busname.c_str(), connection_options );
			return NULL;
		 }
		Log(LOG_WARNING, ":connectBus(%p,%s,%i) : Bus is missing, creating..", handler, busname.c_str(), connection_options );
		WRESULT ret = registerBus_( new BusRing(busname) );
		this->busses.transactionEnd();
		if ( WFAILED(ret) ) {
			Log(LOG_CRITICAL, ":connectBus(%p,%s,%i) : FAILED: Unable to create Bus!", handler, busname.c_str(), connection_options );
			return NULL;
		 }
		connection_options &= ~(Bus::BUS_CREATE_IF_MISSING); // recursive once
		return connectBusEx(handler, busname, connection_options);
	 }
	else {
		bool deferred = ( (connection_options & (Bus::ACTION_DEFERRED)) == Bus::ACTION_DEFERRED);
		bus->addBusListener(handler, deferred);
		this->busses.transactionEnd();
		Log(LOG_VERBOSE, ":connectBus(%p,%s,%i) : Listening to Bus#%ld", handler, busname.c_str(), connection_options, bus->getBusID() );
	 }
	return bus;
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////////////////////

WRESULT BusManager::postMessage( Message* message, const string& busname ) {
	if ( message == NULL ) return NULL;
	MutexLockerRead mL(this->busses.getMutex());
	Bus* bus = this->busses.find(busname);
	if ( bus == NULL ) return WRET_ERR_PARAM;
	bus->postMessage(message);
	return WRET_OK;
 }

////////////////////////////////////////////////////////////////////////////////////////////// BUS OPERATIONS
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////// BUS OPERATIONS

WRESULT BusManager::stopBusses( unsigned long synch_timeout ) {
	WRESULT ret = WRET_OK;
	this->busses.transactionBeginRead();
	Log(LOG_INFO, ":stopBusses() Stopping %d busses..", this->busses.size() );
	tStrBusMap::Iterator it;
	const tStrBusMap::ConstIterator it_end = this->busses.end();
	for ( it=this->busses.begin(); it!=it_end; ++it ) {
		if ( it->second == NULL ) continue;
		if ( it->second->isBusRunning() )
			ret += it->second->stopBus(synch_timeout);
	 }
	this->busses.transactionEnd();
	return ret;
 }

////
//////////////////////////////////////////////////////////////////////////////////////////////////////////////

WRESULT BusManager::createBus( const std::string& bus_name, const std::string& bus_type ) {
	if ( bus_name.size() < 3 ) return WRET_ERR_PARAM;
	MutexLockerWrite mL(this->busses.getMutex());
	bool found = this->busses.exists( bus_name );
	if ( found ) {
		Log(LOG_CRITICAL, ":createBus(%s) : FAILED Bus already exists!", bus_name.c_str() );
		return WRET_ERR_PARAM;
	 }
	Bus* bus = NULL;
	std::string bus_classname = bus_type;
	if ( bus_type.empty() ) bus_classname = BusRing::className();
	Log(LOG_VERBOSE, ":createBus(%s) : Requesting ObjectFactory to allocate [%s]..", bus_name.c_str(), bus_type.c_str() );
	bus = ObjectFactory::createTypeOf<Bus>(bus_classname, 0.0);
	//database->setPermission( Permission(username, group, Permission::RWX, Permission::RX, Permission::Read) );
	if ( bus == NULL ) {
		Log(LOG_CRITICAL, ":createBus(%s) : INTERNAL ERROR Creating Bus! [%s]", bus_name.c_str(), bus_type.c_str() );
		return WRET_ERR_INTERNAL;
	 }
	bus->setName(bus_name);
	return registerBus_( bus );
 }

////////////////////////////////////////////////////////////////////////////////////////////// BUS OPERATIONS
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
///////////////////////////////////////////////////////////////////////////////////////////////////// METHODS

Response* BusManager::mmDoCreateBus( const Request* request ) {
	if ( request == NULL ) return NULL;
	Variant dBusName;
	Variant dBusType;
	request->extractArguments(dBusName, dBusType);
	if ( !dBusName.isStringNotEmpty() )
		return request->replyResponse(WRET_ERR_PARAM, "invalid bus_name");
	if ( !dBusType.isStringNotEmpty() )
		dBusType.set("");
	Log(LOG_VERBOSE, ":mmDoCreateBus(%s,%s)", dBusName.asString().c_str(), dBusType.asString().c_str() );
	WRESULT ret = createBus( dBusName.asString(), dBusType.asString() );
	return request->replyResponse(ret);
 }

Response* BusManager::mmDoControlBus( const Request* request ) {
	if ( request == NULL ) return NULL;
	Variant dBusName; Variant dBusAction;
	request->extractArguments(dBusName, dBusAction);
	if ( !dBusName.isStringNotEmpty() )
		return request->replyResponse(WRET_ERR_PARAM, "invalid bus_name");
	this->busses.transactionBeginRead();
	Bus* bus = this->busses.find(dBusName.asString());
	if ( bus == NULL ) {
		this->busses.transactionEnd();
		return request->replyResponse(WRET_ERR_PARAM, "invalid bus_name");
	 }
	WRESULT ret = WRET_ERR_INTERNAL;
	if ( dBusAction == (int)Bus::STATE_STARTED || dBusAction == "start" ) {
		if ( !bus->isBusRunning() )
			ret = bus->startBus(0); // asynch
	 }
	else if ( dBusAction == (int)Bus::STATE_STOPPED || dBusAction == "stop" ) {
		if ( bus->isBusRunning() )
			ret = bus->stopBus(0); // asynch
	 }
	else if ( dBusAction == "flush" ) {
		bus->flushMessages();  ret = WRET_OK;
	 }
	else if ( dBusAction == "+network" ) {
		bus->sharedNetwork = true; ret = WRET_OK;
	 }
	else if ( dBusAction == "-network" ) {
		bus->sharedNetwork = false; ret = WRET_OK;
	 }
	else if ( dBusAction == "+local" ) {
		bus->sharedLocal = true; ret = WRET_OK;
	 }
	else if ( dBusAction == "-local" ) {
		bus->sharedLocal = false; ret = WRET_OK;
	 }
	this->busses.transactionEnd();
	return request->replyResponse(ret);
 }

///////////////////////////////////////////////////////////////////////////////////////////////////// METHODS
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////////////////////

void BusManager::push_message( Message* message, NetworkChannel* ) {
	if ( message == NULL ) return;
	BusManager* instance = BusManager::getInstance();
	if ( instance == NULL ) return;
	///@todo add security check!
	MutexLockerRead mL(instance->busses.getMutex());
	if ( message->getDestinationBus() == _Bus_ANY ) {
		tStrBusMap::ConstIterator it;
		const tStrBusMap::ConstIterator it_end = instance->busses.end();
		for ( it=instance->busses.begin(); it!=it_end; ++it ) {
			if ( it->second == NULL ) continue;
			if ( !it->second->isSharedNetwork() ) {
				instance->Log(LOG_WARNING, ":push_message(%"PRId64") : Ignoring message for not-network-shared Bus '%s'!", message->getID(), message->getDestinationBus().c_str() );
				continue;
			 }
			it->second->postMessage(message->clone());
		 }
		delete message;
		return;
	 }
	Bus* bus = instance->busses.find(message->getDestinationBus());
	if ( bus == NULL ) {
		instance->Log(LOG_WARNING, ":push_message(%"PRId64") : Ignoring message for missing Bus '%s'!", message->getID(), message->getDestinationBus().c_str() );
		delete message;
		return;
	 }
	if ( !bus->isSharedNetwork() ) {
		instance->Log(LOG_WARNING, ":push_message(%"PRId64") : Ignoring message for not-network-shared Bus '%s'!", message->getID(), message->getDestinationBus().c_str() );
		delete message;
		return;
	 }
	bus->postMessage(message);
 }

void BusManager::push_message( const Message* message, Bus* source ) {
	if ( message == NULL || source == NULL ) return;
	if ( message->getDestinationBus() == source->getName() ) {
		// message reached destination bus, if network is local (private) we do nothing
		if ( !source->isSharedNetwork() ) return;
		// if destination is local kernel, we are done
		if ( message->getDestination().isKernelEmptyOrLocal() ) return;
		// we forward message to Network handler
		NetworkManager::push_message( message, source );
	 }
	else {
		// message need to be routed to destination bus.
		BusManager* instance = BusManager::getInstance();
		if ( instance == NULL ) return;
		if ( message->getDestinationBus() == _Bus_ANY ) {
			// ignore messages from bus!=source, because all shared busses will push as well
			if ( message->getSourceBus() != source->getName() )	return;
		 }
		if ( source->isSharedLocal() ) {
			instance->Log(LOG_DEBUG, "push_message(%"PRId64") Routing message to busses [%s]..", message->getID(), message->getDestinationBus().c_str() );
			// forward message to destination bus (locally!)
			if ( message->getDestination().isKernelEmptyOrLocal() || message->getDestination().isKernelAny() ) {
				MutexLockerRead mL(instance->busses.getMutex());
				tStrBusMap::ConstIterator it;
				const tStrBusMap::ConstIterator it_end = instance->busses.end();
				for ( it=instance->busses.begin(); it!=it_end; ++it ) {
					if ( it->second == NULL ) continue;
					if ( !it->second->isSharedLocal() ) continue; // don't send to not-shared bus!
					if ( it->second == source || message->getSourceBus() == it->second->getName() ) continue; // don't send to source!
					if ( message->getDestinationBus() == _Bus_ANY || message->getDestinationBus() == it->second->getName() )
						it->second->postMessage(message->clone());
				 }
			 }
		 }
		if ( source->isSharedNetwork() ) {
			if ( !message->getDestination().isKernelEmptyOrLocal() || message->getDestination().isKernelAny() ) {
				// we forward message to Network handler
				NetworkManager::push_message( message, source );
			 }
		 }
	 }
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////
/////////////////////////////////////////////////////////////////////////////////////////////// THREAD EVENTS

void BusManager::mbus_event( Bus::BUS_STATE state, Bus* mbus ) {
	if ( mbus == NULL ) return;
	switch( state ) {
		case Bus::STATE_CREATED: {
			break;
		 }
		case Bus::STATE_DESTROYED: {
			// note:
			// Bus is DESTROYING! we cannot access anything uppder-class than wosh::Bus, this includes Object::getName()!
	//		this->busses.transactionBeginWrite();
	//		unRegisterBus_( mbus );
	//		this->busses.transactionEnd();
			break;
		 }
		default: {	Log(LOG_WARNING, ":mbus_event(%d) : Unknown event!", state );
			break;
		 }
	 }
 }

/////////////////////////////////////////////////////////////////////////////////////////////// THREAD EVENTS
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////////////////////

//////////////////////////////////////////////////////////////////////////////////////////////////////////////

}; // namespace wosh
